#ifdef __cplusplus
extern "C"{
#endif

#include "xs.h"
#include "xs_log.h"
#include "xs_io.h"
#include "xs_linux.h"

#ifdef WIN32
int xs_socketpair(int af, int type, int protocol, XSOCKET socks[])
{
    protocol = protocol;
    XSOCKET listener, connector, acceptor;
    struct sockaddr_in listen_addr, connect_addr;
    int size = sizeof(connect_addr);

    listener = socket(af, type, 0);
    if(listener == (XSOCKET)SOCKET_ERROR)
        return -1;

    memset(&listen_addr, 0, sizeof(listen_addr));
    listen_addr.sin_family = AF_INET;
    listen_addr.sin_addr.s_addr = htonl(INADDR_LOOPBACK);
    listen_addr.sin_port = 0;

    if(bind(listener, (struct sockaddr*)&listen_addr, sizeof(listen_addr)) == -1)
    {
        xs_close_socket(listener);
        return -1;
    }

    if(listen(listener, 1) == -1)
    {
        xs_close_socket(listener);
        return -1;
    }


    if(getsockname(listener, (struct sockaddr*)&connect_addr, &size) == -1)
    {
        xs_close_socket(listener);
        return -1;
    }
    connector = socket(af, type, 0);
    if(connector == (XSOCKET)SOCKET_ERROR)
    {
        xs_close_socket(listener);
        return -1;
    }

    if(connect(connector, (struct sockaddr*)&connect_addr, size) == -1)
    {
        xs_close_socket(connector);
        xs_close_socket(listener);
        return -1;
    }

    size = sizeof(listen_addr);
    acceptor = accept(listener, (struct sockaddr*)&listen_addr, &size);
    if(acceptor == (XSOCKET)SOCKET_ERROR)
    {
        xs_close_socket(connector);
        xs_close_socket(listener);
        return -1;
    }

    socks[0] = connector;
    socks[1] = acceptor;

    xs_logd("socketpair=%d, %d", socks[0], socks[1]);

    return 0;
}

int xs_inet_pton(int af, const char *src, void *dst)
{
    if(af == AF_INET6)
        assert(0);

    struct	in_addr* addr = (struct in_addr*)dst;
    addr->S_un.S_addr = inet_addr(src);

    return 0;
}

/* use select do epoll */
typedef struct xs_epoll_event_t
{
    XSOCKET fd;
    struct epoll_event ev;
    xs_list_t lnode;
    struct rb_node rb;
} xs_epoll_event_t;

typedef struct xs_epoll_evs_t
{
    xs_list_t ready;
    xs_list_t go;
    struct rb_root tree;
} xs_epoll_evs_t;

pthread_mutex_t g_epoll_mutex = PTHREAD_MUTEX_INITIALIZER;
xs_epoll_evs_t* g_epoll_evs = NULL;
#define xs_epoll_lock() pthread_mutex_lock(&g_epoll_mutex)
#define xs_epoll_unlock() pthread_mutex_unlock(&g_epoll_mutex)
#define xs_epoll_find(_tree, _key) rb_find(_tree, (const void*)(intptr_t)_key)

int xs_epoll_evs_cmp(const struct rb_node* node, const void* ptr)
{
    xs_epoll_event_t* ev = rb_entry(node, xs_epoll_event_t, rb);
 //   xs_logd("%d, key=%d", ev->fd, (XSOCKET)(intptr_t)ptr);
    return ev->fd - (XSOCKET)(intptr_t)ptr;
}

xs_epoll_event_t* xs_epoll_event_create(XSOCKET fd, struct epoll_event* ev)
{
    xs_epoll_event_t* xsev = xs_malloc(sizeof(xs_epoll_event_t));
    memcpy(&xsev->ev, ev, sizeof(*ev));
    xsev->fd = fd;
    rb_init_node(&xsev->rb);
    xs_list_init(&xsev->lnode);
    return xsev;
}

int xs_epoll_ctl(int epollfd, int method, XSOCKET fd, struct epoll_event* ptr)
{
    int ret = 0;
    struct rb_node* node;
    xs_epoll_event_t* xsev;

    epollfd = epollfd;

    xs_epoll_lock();
    if(g_epoll_evs == NULL)
    {
        g_epoll_evs = xs_malloc(sizeof(xs_epoll_evs_t));
        xs_list_init(&g_epoll_evs->go);
        xs_list_init(&g_epoll_evs->ready);
        rb_init(&g_epoll_evs->tree, xs_epoll_evs_cmp);
    }
    node = xs_epoll_find(&g_epoll_evs->tree, fd);

    if(node)
    {
        xsev = xs_entry(node, xs_epoll_event_t, rb);
        switch (method)
        {
            case EPOLL_CTL_DEL:
                rb_erase(&xsev->rb, &g_epoll_evs->tree);
                xs_list_del(&xsev->lnode);
                goto RET;

            case EPOLL_CTL_MOD:
                memcpy(&xsev->ev, ptr, sizeof(*ptr));
                xs_list_move_tail(&g_epoll_evs->ready, &xsev->lnode);
                goto RET;

            case EPOLL_CTL_ADD:
                /* already exist, can not add again */
                xs_logd("already exist fd=%d", xsev->fd);
                ret = -1;
                goto RET;
        }
    }
    else
    {
        switch (method)
        {
            /* not exist, can not del and mod */
            case EPOLL_CTL_DEL:
            case EPOLL_CTL_MOD:
                ret = -1;
                goto RET;
            case EPOLL_CTL_ADD:
                /* add the event in ready list */
                xsev = xs_epoll_event_create(fd, ptr);
                rb_insert(&xsev->rb, &g_epoll_evs->tree, (void*)(intptr_t)fd);
                xs_list_add_tail(&g_epoll_evs->ready, &xsev->lnode);
                goto RET;
        }
    }
RET:
    xs_epoll_unlock();
    return ret;
}

int xs_epoll_wait(int epollfd, struct epoll_event* ev_ret, int count, int t)
{
    int i;
    xs_list_t* node;
    xs_epoll_event_t* ev;
    struct timeval tv;
    fd_set r;
    fd_set w;
    FD_ZERO(&r);
    FD_ZERO(&w);

    xs_epoll_lock();
    if(!xs_list_empty(&g_epoll_evs->go))
        goto RET;

    epollfd = epollfd;
    node = g_epoll_evs->ready.next;
    while(node != &g_epoll_evs->ready)
    {
        ev = xs_entry(node, xs_epoll_event_t, lnode);
        node = node->next;
        if(ev->ev.events & EPOLLIN)
        {
            FD_SET(ev->fd, &r);
        }
        if(ev->ev.events & EPOLLOUT)
        {
            FD_SET(ev->fd, &w);
        }
    } 
    xs_epoll_unlock();


    tv.tv_sec = t/1000;
    tv.tv_usec = t%1000*1000;

    int ret = select(1024, &r, &w, NULL, &tv);
    if(ret <= 0)
        return ret;

    xs_epoll_lock();

    if(!xs_list_empty(&g_epoll_evs->ready))
    {
        node = g_epoll_evs->ready.next;
        while(node != &g_epoll_evs->ready)
        {
            ev = xs_list_entry(node, xs_epoll_event_t, lnode);
            node = node->next;
            if(FD_ISSET(ev->fd, &r) || FD_ISSET(ev->fd, &w))
            {
                xs_list_move_tail(&g_epoll_evs->go, &ev->lnode);
            }
        }
    }

RET:
    for(i=0; i<count; ++i)
    {
        if(xs_list_empty(&g_epoll_evs->go))
            break;
        
        ev = xs_list_first_entry(&g_epoll_evs->go, xs_epoll_event_t, lnode);
        xs_list_del(&ev->lnode);
        memcpy(&ev_ret[i], &ev->ev, sizeof(struct epoll_event));
    }
    xs_epoll_unlock();
    return i;
}

int xs_epoll_create(int __c)
{
    g_epoll_evs = xs_malloc(sizeof(xs_epoll_evs_t));
    xs_list_init(&g_epoll_evs->go);
    xs_list_init(&g_epoll_evs->ready);
    rb_init(&g_epoll_evs->tree, xs_epoll_evs_cmp);
    __c = __c
    return 0;
}

int xs_epoll_close(int fd)
{
    fd = fd;
    return 0;
}

#else		/* End ifdef __WIN32 and Start of Linux */

#include "xs_epoll.h"
int xs_inet_pton(int __a, const char * __s, void * __d)
{
	return (inet_pton(__a, __s, __d));
}
int xs_socketpair(int __a, int __type, int __pro, XSOCKET __socks [ 2 ])
{
	int __result__ = -1;

	__result__ = (socketpair(__a, __type, __pro, __socks));
	if(__result__ < 0){
		xs_dump_error_str(errno);
		return EXIT_ERROR;
	}
	
	return EXIT_OK;
}
int xs_epoll_create(int __c)
{
	XSOCKET __efd__ = -1;

	__efd__ = epoll_create(__c);
	if(__efd__ < 0){
		xs_dump_error_str(errno);
		return EXIT_ERROR;
	}
		
	return (__efd__);
}
int xs_epoll_ctl(XSOCKET __efd, int __method, XSOCKET __fd, struct epoll_event* __p)
{
	int __result__ = -1;

	__result__ = (epoll_ctl(__efd, __method, __fd, __p) < 0);
	if(__result__ < 0){
		xs_dump_error_str(errno);
		return EXIT_ERROR;
	}
	
	return EXIT_OK;
}

int xs_epoll_wait(XSOCKET __efd, struct epoll_event* __ev, int __c, int __tw)
{
	int __num__ = -1;

	__num__ = epoll_wait(__efd, __ev, __c, __tw);
	if(__num__ < 0)
		__num__ = EXIT_ERROR;

	return (__num__);
}

int xs_epoll_close(XSOCKET __f)
{
	return xs_release_socket(__f);
}

#endif


#include "xs_network.h"
typedef struct block_queue  
{  
    int queue[xs_defult_thread_size];  
    long size;  
    pthread_cond_t cond;  
    pthread_mutex_t mutex;  
}block_queue_t;  
block_queue_t bq;  

typedef struct block_queue_param  
{  
    void* func;  
    void* queue;  	/* Point to block queue structure */
}block_queue_param_t;  
block_queue_param_t bqp; 

int g_xs_thread_count = xs_defult_thread_size;
#define BUFFER_SIZE 		1024
#include <sys/resource.h>
int g_epoll_fd = -1;
xs_epoll_event xs_ev, xs_events[EPOLL_DEFAULT_SIZE];
int g_epoll_size = EPOLL_DEFAULT_SIZE;
int g_serv_fd = -1;

void xs_network_epoll_echo(void* data)
{
      int socket = *(int *)data;
	xs_logd("%d !\n", socket);
	char buffer[BUFFER_SIZE];
	
	//while(1)
	{
		/* We only send what recevied just now */
		int   length = xs_net_recv(socket, buffer, BUFFER_SIZE);
		if(length){
			xs_net_send(socket, buffer, strlen(buffer));
			memset(buffer, 0, BUFFER_SIZE);
		}
	}
}

void *xs_handle_queue(void *param)  
{  
	void(* func)(void* );  
	int fd, i;  

	block_queue_t* bque = ((block_queue_param_t*)param)->queue;  
	func = ((block_queue_param_t*)param)->func; 

	xs_pthread_cond_init(&bque->cond,  NULL); 
	xs_pthread_mutex_init(&bque->mutex,  NULL);
	
	for(;;)  
	{  
		if(xs_pthread_mutex_lock(&bque->mutex) == EXIT_OK) {
			
			xs_pthread_cond_wait(&bque->cond, &bque->mutex);  
			if(bque->size==0)  
			{  
				xs_pthread_mutex_unlock(&bque->mutex);
				continue;
			}
			else  
			{  				

				fd = bque->queue[0];  

				for(i = 0; i < bque->size - 1; ++i)  
					bque->queue[i] = bque->queue[i + 1];  
				
				bque->queue[bque->size-1] = 0;  
				bque->size--;  
			}  
		    	xs_pthread_mutex_unlock(&bque->mutex);  
		}
		
		func((void *)&fd);  
	}  
}  

int xs_init_threads( void )  
{  
	int i = 0, ret;  

	bqp.func = (void*)xs_network_epoll_echo; 
	bqp.queue = (void *)&bq;

	xs_pthread_cond_init(&bq.cond,NULL);  
	xs_pthread_mutex_init(&bq.mutex,NULL);  
	xs_pthread_t child_thread[g_xs_thread_count];  
	xs_pthread_attr_t child_thread_attr[g_xs_thread_count];  
   
	for( i = 0; i < g_xs_thread_count; ++i)  {  
		ret = pthread_attr_init(&child_thread_attr[i]);  
		if(ret != 0) xs_logd("error to init attr !\n");
		pthread_attr_setdetachstate(&child_thread_attr[i], PTHREAD_CREATE_DETACHED);  
		if( pthread_create(&child_thread[i],
			&child_thread_attr[i], xs_handle_queue, (void *)&bqp) < 0 )  {  
				xs_logd("pthread_create Failed : %s - %m\n",strerror(errno));  
				return 1;  
		}  
    	}  
  	return 0;
}  
int xs_init_server(short int port, const char *name)
{
	struct sockaddr_in *server_addr;  
	server_addr = malloc(sizeof(struct sockaddr_in));  
	server_addr->sin_family = AF_INET;  

	if(name == NULL)	
		server_addr->sin_addr.s_addr = htons(INADDR_ANY); 
	else
		server_addr->sin_addr = xs_sock_getip(name);
	server_addr->sin_port = htons(port);  
	int server_socket = xs_net_socket(AF_INET, SOCK_STREAM, 0);  
	set_nonblocking(server_socket);  
	xs_net_bind(server_socket,(struct sockaddr*)server_addr,sizeof(struct sockaddr_in));
	xs_net_listen(server_socket, xs_default_listen_size); 

	struct rlimit rt;  
	rt.rlim_max = rt.rlim_cur = g_epoll_size;  
	if (setrlimit(RLIMIT_NOFILE, &rt) == -1) {  
		xs_logd("setrlimit - %m");  
		exit(1);  
	}  

	return server_socket;  
}

static void xs_insert_queue(struct block_queue *bque, int fd)  
{  
	xs_pthread_mutex_lock(&bque->mutex);  

	if(bque->size == THREADS_COUNT)  
		return;  
	bque->queue[bque->size] = fd;  
	if(++bque->size > THREADS_COUNT)  {  
		fprintf(stderr,"Queue size over folow.%ld", bque->size);  
		exit (1);  
	}  
	xs_pthread_cond_signal(&bque->cond);  
	xs_pthread_mutex_unlock(&bque->mutex);  

	int i = 0;
	printf("dump fd pool:");
	
	for(i = 0; i < bque->size; i++)
		printf("%d ", bque->queue[i]);
	printf("\n");
} 
static inline void xs_handler(void* fd)  
{  
    printf("handler:fd => %d\n", *(int *)(fd));  
    xs_insert_queue(&bq, *(int *)(fd));  
}  

int xs_epoll_main()
{
	int nfds, n;
	
	g_serv_fd = xs_init_server(11111, NULL);
	xs_logd("server thread [FD:%d] is ready for ...", g_serv_fd);

	 if(xs_init_threads() == 0)  
        	xs_logd("Threads ready to use !");  
 
	g_epoll_fd = xs_epoll_create(g_epoll_size);

	if(g_epoll_fd < 0) return 0;

	xs_logd("epoll create success -> fd : %d", g_epoll_fd);
	
	xs_ev.events = EPOLLIN | EPOLLOUT|EPOLLET;
	xs_ev.data.fd = g_serv_fd;

	if (xs_epoll_ctl(g_epoll_fd, EPOLL_CTL_ADD, g_serv_fd, &xs_ev) < 0) {  
		
	        fprintf(stderr, "epoll set insertion error: fd=%d < 0",  
	                g_serv_fd);  
	        return -1;  
    	}
	
	 for(;;) {  
	        struct sockaddr_in local;  
	        socklen_t length = sizeof(local);  
	        int client = -1;  
	          
	        nfds = xs_epoll_wait(g_epoll_fd, xs_events, EPOLL_DEFAULT_SIZE, -1);  

	        for(n = 0; n < nfds; ++n) {  

			if(xs_events[n].data.fd == g_serv_fd) {  
				client = accept(g_serv_fd, (struct sockaddr *) &local,&length);  
				if(client < 0){  
					xs_logd("%s", strerror(errno));  
					continue;  
				}  
				else
				{
					xs_logd("add socket pool : %d", client);

					set_nonblocking(client);  
					xs_ev.events = EPOLLIN | EPOLLOUT|EPOLLET;  
					xs_ev.data.fd = client;  

					if (xs_epoll_ctl(g_epoll_fd, EPOLL_CTL_ADD, client, &xs_ev) < 0) {  
						fprintf(stderr, "epoll set insertion error: fd=%d < 0",  client);  
						//return -1;
						continue;
					}  
					client = -1;
				}
			}  
			else  /* It's a client fd that needed to process */
				xs_handler((void *)&xs_events[n].data.fd);
	        }  
    	}  

	xs_close_socket(g_serv_fd); 
	xs_epoll_close(g_epoll_fd);

}

#ifdef __cplusplus
}
#endif
